package com.fitz.remote.client;

import com.fitz.invoke.RpcInvoker;
import com.fitz.registry.InstanceInfo;
import com.fitz.remote.codec.RpcDecoder;
import com.fitz.remote.codec.RpcEncoder;
import com.fitz.remote.handler.RpcClientHandler;
import com.fitz.remote.model.BasicObject;
import com.fitz.remote.model.RpcRequest;
import com.fitz.remote.model.RpcResponse;
import com.fitz.serializer.SerializerType;
import com.fitz.util.IpUtil;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author 2020-5-14 14:58
 */

public class ConsumerClient {

    private static final Logger logger = LoggerFactory.getLogger(ConsumerClient.class);
    private Bootstrap boot = new Bootstrap();

//    private String serverAddress;

    private InstanceInfo instance;

    private Class<? extends BasicObject> clazz;

    private byte serializerType;

    private RpcClientHandler rpcClientHandler;

    private Channel channel;

    public Channel getChannel() {
        return channel;
    }

    private volatile boolean isOpen;

    public boolean isOpen() {
        return isOpen;
    }

    public void open(){
        this.isOpen = true;
    }

    public void close(){
        this.isOpen = false;
    }

//    public String getServerAddress() {
//        return serverAddress;
//    }
//
//    public void setServerAddress(String serverAddress) {
//        this.serverAddress = serverAddress;
//    }


    public InstanceInfo getInstance() {
        return instance;
    }

    public void setInstance(InstanceInfo instance) {
        this.instance = instance;
    }

    public ConsumerClient(InstanceInfo instance) {
        this.instance = instance;
        this.clazz = RpcRequest.class;
        this.serializerType = instance.getSerializerType();
        initClient();
    }

    private void initClient(){
        rpcClientHandler = new RpcClientHandler();
        String os = System.getProperty("os.name");
        logger.info("System:[{}]",os);
        EventLoopGroup group = null;
        Class<? extends Channel> channelClass = null;
        if (os.toLowerCase().startsWith("win")||os.toLowerCase().startsWith("mac")) {
            group = new NioEventLoopGroup();
            channelClass = NioSocketChannel.class;
        }else {
            group = new EpollEventLoopGroup();
            channelClass = EpollSocketChannel.class;
        }
        //设置channel实现类，为channel注册处理所有io事件模型
        boot.group(group)
                .channel(channelClass)
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .option(ChannelOption.TCP_NODELAY, true);
        boot.handler(initHandler());
    }

    private ChannelInitializer initHandler(){
        return new ChannelInitializer() {
            @Override
            protected void initChannel(Channel channel) throws Exception {
                channel.pipeline()
                        .addFirst("idleStateHandler", new IdleStateHandler(60, 0, 0))
                        .addLast("RpcDecoder",new RpcDecoder(serializerType, RpcResponse.class))
                        .addLast("RpcEncoder",new RpcEncoder(serializerType))
                        .addLast(rpcClientHandler);

            }
        };
    }

    public ChannelFuture connect(){
        ChannelFuture channelFuture = null;
        try {
            channelFuture = boot.connect(instance.getHost(),instance.getPort()).sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        if (channelFuture.isSuccess()){
            channel = channelFuture.channel();
            logger.info("连接服务端成功");
        }else {
            channelFuture.cause().printStackTrace();
            logger.error("连接服务端失败");
        }
        return channelFuture;
    }

    public void sendRequest(RpcRequest rpcRequest){
        channel.writeAndFlush(rpcRequest);
    }
}
